package com.csdn.eval.spring.mqtt.config;

import com.csdn.eval.spring.mqtt.constants.ChannelName;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.expression.FunctionExpression;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import java.util.UUID;

@Configuration
public class EvalSpringMqttConfiguration {

    @Autowired
    private EvalSpringMqttProperties evalSpringMqttProperties;

    private MqttPahoMessageDrivenChannelAdapter createAdapter(String[] topics, String clientId, MqttPahoClientFactory factory, MessageChannel channel) {
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(clientId, factory, topics);
        adapter.setConverter(createMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(channel);
        return adapter;
    }

    private DefaultPahoMessageConverter createMessageConverter() {
        DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter();
        converter.setPayloadAsBytes(true);
        return converter;
    }

    /**
     * 订阅适配器
     */
    @Bean
    public MqttPahoMessageDrivenChannelAdapter topicAdapter(@Qualifier(ChannelName.INBOUND) MessageChannel channel) {
        return createAdapter(evalSpringMqttProperties.getTopic().split(","), evalSpringMqttProperties.getClientId()
                + UUID.randomUUID(), publicMqttClientFactory(), channel);
    }


    private MqttPahoClientFactory createClientFactory(EvalSpringMqttProperties config) {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[]{"tcp://" + config.getHost() + ":" + config.getPort()});
        options.setUserName(config.getUsername());
        options.setPassword(config.getPassword().toCharArray());
        options.setConnectionTimeout(config.getConnectionTimeout());
        options.setKeepAliveInterval(config.getKeepAliveInterval());
        options.setCleanSession(config.isCleanSession());
        options.setAutomaticReconnect(true);
        factory.setConnectionOptions(options);
        return factory;
    }

    public MqttPahoClientFactory publicMqttClientFactory() {
        return createClientFactory(evalSpringMqttProperties);
    }

    private MessageHandler createMessageHandler(String clientIdPrefix, MqttPahoClientFactory factory) {
        MqttPahoMessageHandler handler = new MqttPahoMessageHandler(clientIdPrefix, factory);
        handler.setAsync(false);
        handler.setDefaultQos(0);
        handler.setConverter(createMessageConverter());
        handler.setTopicExpression(new FunctionExpression<Message<?>>(m ->
                m.getHeaders().getOrDefault(
                        MqttHeaders.TOPIC,
                        "fallback/topic" // 如果消息头没有 Topic，则使用默认值
                )
        ));
        return handler;
    }

    /**
     * 出站適配器，用于发送消息
     */
    @Bean
    @ServiceActivator(inputChannel = ChannelName.OUTBOUND)
    public MessageHandler mqttOutbound() {
        return createMessageHandler(evalSpringMqttProperties.getClientId() +
                UUID.randomUUID(), publicMqttClientFactory());
    }


}